﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;

using WorkStation.Frameworks.EasyBus;
using WorkStation.Frameworks.EasyBus.Handles;
using WorkStation.Frameworks.EasyBus.Pipeline;
using WorkStation.Frameworks.EasyBus.Pipeline.Extensions;
using WorkStation.Frameworks.EasyBus.Pipeline.Handles;
using WorkStation.Frameworks.EasyBus.Extensions;
using WorkStation.Frameworks.EasyBus.DbModel;

using Newtonsoft.Json;

namespace WorkStation.Frameworks.EasyBus.Repositories.ZooKeeper
{
    public static class IBusExtensions
    {
        /// <summary>
        /// 设置消息注册中心仓储组件 ZooKeeper实现
        /// </summary>
        /// <param name="bus"></param>
        /// <param name="connectionString">ZooKeeper字符串</param>
        /// <returns></returns>
        public static IBus UserZooKeeper(this IBus bus, string connectionString, TimeSpan sessionTimeout) 
        {
            RegisterCenterRepository repository = new RegisterCenterRepository(connectionString, bus, sessionTimeout);

            bus.SetRegisterCenterRepository(repository);

            IBusDataManager m = bus as IBusDataManager;

            // 在Bus中注册一个消息订阅变更的处理，重置订阅的处理器，ZooKeeper实现由ZooKeeper接收通知并发布事件
            m.BusEventPipeline.Subscribe<SubscribeInfoChangeEvent>((context, handle) =>
            {
                SubscribeInfoChangeEvent @event = context.Input as SubscribeInfoChangeEvent;

                IBusDataManager manager = (context as BusActionContext).Bus as IBusDataManager;

                // 如果是SubscribeInfoChangeType.OffLine事件，则只会取消订阅，若是新增或者变更会执行下面If里的代码重新注册订阅
                manager.SendMessagePipeline.UnSubscribe(@event.Change.RouteKey, DefaultPublishHandle.GetHandleName(@event.Change));

                List<SubscribeInfo> subs = bus.DataBag["SubscribeInfo"] as List<SubscribeInfo>;

                subs.RemoveAll(v => v.Name == @event.Change.Name && v.NodeName == @event.Change.NodeName && v.BusName == @event.Change.BusName);

                if (@event.SubscribeInfoChangeType == SubscribeInfoChangeTypeEnum.Change || @event.SubscribeInfoChangeType == SubscribeInfoChangeTypeEnum.OnLine) 
                {
                    bus.Subscribe(@event.Change);

                    subs.Add(@event.Change);
                }

                IHandle messageRouteHandle = manager.SendMessagePipeline.GetHandle("System.Object", MessageRouteHandle.NAME);
                if (messageRouteHandle != null)
                {
                    (messageRouteHandle as MessageRouteHandle).Init(manager as IBus);
                }

                return new SuccessHandleResult(context.Input, handle);
            });
            
            // 监听信息变更处理订阅 
            m.BusEventPipeline.Subscribe<ListenerInfoChangeEvent>((context, handle) => 
            {
                IBusDataManager manager = (context as BusActionContext).Bus as IBusDataManager;

                ListenerInfoChangeEvent @event = context.Input as ListenerInfoChangeEvent;

                if (@event.ListenerInfoChangeType == ListenerInfoChangeTypeEnum.Change) 
                {
                    IMessageListener listener = null;

                    if (manager.Server.Listeners.TryGetValue(@event.Change.Name, out listener))
                    {
                        listener.StopListen();

                        listener.ProcessThreadCount = @event.Change.ProcessThreadCount;
                        listener.Enabled = @event.Change.Enabled;

                        listener.BeginListen();
                    }
                }

                return new SuccessHandleResult(context.Input, handle);
            });

            return bus;
        }
    }
}
